package cn.gwm.flink.streaming.function.apply;

import cn.hutool.json.JSONObject;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.util.Collector;

import java.util.Iterator;

/**
 * @ClassName OdsCanGpsCoGroupFunction
 * @Author xzh
 * @Date 2022/12/7 16:17
 **/
public class OdsCanGpsCoGroupFunction   implements CoGroupFunction<JSONObject,JSONObject,JSONObject> {
    @Override
    public void coGroup(Iterable<JSONObject> first, Iterable<JSONObject> second, Collector<JSONObject> collector) throws Exception {
        Iterator<JSONObject> can = first.iterator();
        while(can.hasNext()) {
            JSONObject canObj = can.next();
            Iterator<JSONObject> gps = second.iterator();
            while(gps.hasNext()) {
                JSONObject gpsObj = gps.next();
                if (gpsObj != null) {
                    canObj.set("tboxMoniGPSstatus", gpsObj.getStr("tboxMoniGPSstatus"));
                    canObj.set("tboxMoniLongitude", gpsObj.getStr("tboxMoniLongitude"));
                    canObj.set("tboxMoniLatitude", gpsObj.getStr("tboxMoniLatitude"));
                    canObj.set("imsi2", gpsObj.getStr("imsi"));
                    canObj.set("command2", gpsObj.getStr("command"));
                    canObj.set("user_id2", gpsObj.getStr("user_id"));

                }
            }
            collector.collect(canObj);
        }
    }
}
